package cn.kgc.gmall.app.dws;

import cn.kgc.gmall.bean.VisitorStats;
import cn.kgc.gmall.utils.ClickhouseUtil;
import cn.kgc.gmall.utils.DateTimeUtil;
import cn.kgc.gmall.utils.MyKafkaUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.time.Duration;
import java.util.Date;

/**
 * 将三条流中的数据进行union合并
 * 水位线设置
 * 分组
 * 开窗
 * 聚合
 * 写出到clickHouse
 */
public class VisitorStatsApp {
    public static void main(String[] args) throws Exception {
        // 模板代码 运行环境  检查点
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        // 2.检查点相关设置
        env.enableCheckpointing(5*1000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        env.getCheckpointConfig()
                .enableExternalizedCheckpoints(
                        CheckpointConfig
                                .ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
                );
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/gmall/flink/checkpoint"));
        System.setProperty("HADOOP_USER_NAME","atkgc");

        String groupId = "visitor_stats_app";

        //TODO 1.从Kafka的pv、uv、跳转明细主题中获取数据
        String pageViewSourceTopic = "dwd_topic_page";
        String uniqueVisitSourceTopic = "dwm_unique_visit";
        String userJumpDetailSourceTopic = "dwm_user_jump_detail";

        // TODO 2.读取Kafka中的数据到流中

        DataStreamSource<String> pageViewSource = env.addSource(MyKafkaUtils.getKafkaConsumerSource(pageViewSourceTopic, groupId));
        DataStreamSource<String> uniqueVisitSource = env.addSource(MyKafkaUtils.getKafkaConsumerSource(uniqueVisitSourceTopic, groupId));
        DataStreamSource<String> userJumpDetailSource = env.addSource(MyKafkaUtils.getKafkaConsumerSource(userJumpDetailSourceTopic, groupId));

        // 测试展示数据
        //pageViewSource.print("pageView:");
        //uniqueVisitSource.print("uniqueVisit");
        //userJumpDetailSource.print("userJumpDetail");

        // TODO 3.将数据转化为jsonObject 封装到VisitorStats
        // 页面数据
        SingleOutputStreamOperator<VisitorStats> pageViewMap = pageViewSource.map(new MapFunction<String, VisitorStats>() {
            @Override
            public VisitorStats map(String value) throws Exception {
                // 转化为jsonObject
                JSONObject jsonObject = JSON.parseObject(value);
                // 对VisitorStats进行赋值
                JSONObject common = jsonObject.getJSONObject("common");
                String vc = common.getString("vc");
                String ch = common.getString("ch");
                String ar = common.getString("ar");
                String is_new = common.getString("is_new");
                JSONObject page = jsonObject.getJSONObject("page");
                String last_page_id = page.getString("last_page_id");
                // 获取持续访问时间
                Long dur = page.getLong("during_time");
                // 计算进入次数  如果是第一次访问进入次数就是0 如果不是第一次访问进入次数就是1
                Long sv = last_page_id != null && last_page_id.length() > 0 ? 0L : 1L;
                // 获取时间戳
                Long ts = jsonObject.getLong("ts");
                return new VisitorStats(
                        "","",vc,ch,ar,is_new,0L,1L,sv,0L,dur,ts
                );
            }
        });

        // 用户独立访问数
        SingleOutputStreamOperator<VisitorStats> uniqueVisitMap = uniqueVisitSource.map(new MapFunction<String, VisitorStats>() {
            @Override
            public VisitorStats map(String value) throws Exception {
                // 转化为jsonObject
                JSONObject jsonObject = JSON.parseObject(value);
                // 对VisitorStats进行赋值
                JSONObject common = jsonObject.getJSONObject("common");
                String vc = common.getString("vc");
                String ch = common.getString("ch");
                String ar = common.getString("ar");
                String is_new = common.getString("is_new");
                // 获取时间戳
                Long ts = jsonObject.getLong("ts");

                return new VisitorStats(
                        "", "", vc, ch, ar, is_new, 1L, 0L, 0L, 0L, 0L, ts
                );
            }
        });
        // 用户跳出行为
        SingleOutputStreamOperator<VisitorStats> userJumpDetailMap = userJumpDetailSource.map(new MapFunction<String, VisitorStats>() {
            @Override
            public VisitorStats map(String value) throws Exception {
                // 转化为jsonObject
                JSONObject jsonObject = JSON.parseObject(value);
                // 对VisitorStats进行赋值
                JSONObject common = jsonObject.getJSONObject("common");
                String vc = common.getString("vc");
                String ch = common.getString("ch");
                String ar = common.getString("ar");
                String is_new = common.getString("is_new");
                // 获取时间戳
                Long ts = jsonObject.getLong("ts");
                return new VisitorStats(
                        "", "", vc, ch, ar, is_new, 0L, 0L, 0L, 1L, 0L, ts
                );
            }
        });

        // TODO 4.将三条流的数据进行合流操作
        DataStream<VisitorStats> unionVisitorStats = pageViewMap.union(uniqueVisitMap, userJumpDetailMap);

        // TODO 5.设置水位线
        SingleOutputStreamOperator<VisitorStats> visitorStatsWatermark = unionVisitorStats.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(3L))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<VisitorStats>() {
                                    @Override
                                    public long extractTimestamp(VisitorStats visitorStats, long recordTimestamp) {
                                        return visitorStats.getTs();
                                    }
                                }
                        )
        );
        // TODO 6.分组
        // visitorStatsWatermark.print();
        // 按照四个维度进行分组 渠道、新老用户、app版本、省市区域 4元组
        // visitorStatsWatermark.keyBy(x -> Tuple4.of(x.getCh(),x.getIs_new(),x.getVc(),x.getAr()))
        KeyedStream<VisitorStats, Tuple4<String, String, String, String>> visitorStatsKeyBy = visitorStatsWatermark.keyBy(new KeySelector<VisitorStats, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(VisitorStats visitorStats) throws Exception {
                return Tuple4.of(
                        visitorStats.getCh(),
                        visitorStats.getIs_new(),
                        visitorStats.getVc(),
                        visitorStats.getAr()
                );
            }
        });

        // TODO 7.开窗
        WindowedStream<VisitorStats, Tuple4<String, String, String, String>, TimeWindow> visitorStatsWindow = visitorStatsKeyBy.window(
                TumblingEventTimeWindows.of(Time.seconds(10L))
        );
        // 对比

        // TODO 8.聚合
        SingleOutputStreamOperator<VisitorStats> reduceVisitorStats = visitorStatsWindow.reduce(
                new ReduceFunction<VisitorStats>() {
                    @Override
                    public VisitorStats reduce(VisitorStats vs1, VisitorStats vs2) throws Exception {
                        vs1.setUv_ct(vs1.getUv_ct() + vs2.getUv_ct());
                        vs1.setPv_ct(vs1.getPv_ct() + vs2.getPv_ct());
                        vs1.setSv_ct(vs1.getSv_ct() + vs2.getSv_ct());
                        vs1.setUj_ct(vs1.getUj_ct() + vs2.getUj_ct());
                        vs1.setDur_sum(vs1.getDur_sum() + vs2.getDur_sum());
                        return vs1;
                    }
                },
                // 调用底层api
                new ProcessWindowFunction<VisitorStats, VisitorStats, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void process(Tuple4<String, String, String, String> stringStringStringStringTuple4, ProcessWindowFunction<VisitorStats, VisitorStats, Tuple4<String, String, String, String>, TimeWindow>.Context context, Iterable<VisitorStats> elements, Collector<VisitorStats> out) throws Exception {
                        // 获取窗口开始时间
                        long start = context.window().getStart();
                        long end = context.window().getEnd();
                        // 获取窗口结束时间
                        for (VisitorStats visitorStats : elements) {
                            // 时间转化为日期格式字符串
                            visitorStats.setStt(DateTimeUtil.toYMDhms(new Date(start)));
                            visitorStats.setEdt(DateTimeUtil.toYMDhms(new Date(end)));
                            out.collect(visitorStats);
                        }
                    }
                }

        );

        // TODO 9.写出到ClickHouse数据库
        reduceVisitorStats.addSink(


            ClickhouseUtil.getJdbcSink(
                    "insert into visitor_stats_2022 values(?,?,?,?,?,?,?,?,?,?,?,?)"
            )
        );


        env.execute();
    }
}
